
package cn.tang.tframe.springredis.mq;

import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.springredis.ISpringRedisCache;
import cn.tang.tframe.springredis.QtProjectRedisClient;
import cn.tang.tframe.springredis.mq.task.AbstractTask;
import cn.tang.tframe.springredis.mq.task.Tasks;
import cn.tang.tframe.threadpool.QtThreadPoolManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;

import java.util.ArrayList;
import java.util.List;

/**
 * 任务线程的基类.
 * 
 * @author tanghc
 */
public abstract class RedisMqConsumer<T> implements Runnable {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	public RedisMqConsumer() {
		init();
	}

	private void init() {

		try {
			logger.info(" start init task:key={}", getRedisKey());
			QtThreadPoolManager.getInstance().addThread(this);
			logger.info(" end init task succ:key={}", getRedisKey());
		}
		catch (Exception e) {
			logger.error("init task exception:key={}", getRedisKey(), e);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public void run() {

		while (!Thread.currentThread().isInterrupted()) {
			try {
				final String key = getRedisKey();
				logger.info("~~~~~~~~~~~~redis task~~~~~~~~~~~~key:{}", key);
				ApplicationContext context = getContext();
				if (null == context) {
					logger.error(
						"no springcontext for task {}", this.getClass());
					return;
				}
				// 非阻塞方式
				// T object = redisTemplate.boundListOps(key).rightPop();
				// 阻塞方式
				final ISpringRedisCache<Object> springRedisCache =
					QtProjectRedisClient.getSpringRedisCache();
				T object = springRedisCache.execute(new RedisCallback<T>() {

					@Override
					public T doInRedis(RedisConnection conn)
						throws DataAccessException {

						List<byte[]> list = conn.bRPop(10000, key.getBytes());
						if (list == null) {
							logger.debug(
								"redis task~ get nothing in redis queue: key={}",
								key);
							return null;
						}
						/**
						 * 列表第一个为key，第二个为value
						 */
						// final String key = (String)
						// redisTemplate.getKeySerializer().deserialize(list.get(0));
						final T object =
							(T) springRedisCache.getRedisTemplate().getValueSerializer().deserialize(
								list.get(1));
						// logger.debug("key={},object={}",key,object);
						return object;
					}

				});
				if (object != null) {
					logger.debug(
						"redis task~ get object in redis queue: key={},object={}",
						key, object);
					List<Class<? extends AbstractTask>> taskList = getTasks();
					for (Class<? extends AbstractTask> taskClass : taskList) {
						AbstractTask task = context.getBean(taskClass);
						if (null != task) {
							task.setData(object);
							QtThreadPoolManager.getInstance().addThread(task);
						}
					}
				}
				else {
					logger.debug(
						"redis task~ get object null in redis queue: key={}",
						key);
				}
				Thread.sleep(1);
			}
			catch (Throwable e) {
				logger.error("任务线程运行失败: key={}", getRedisKey(), e);
			}
		}
	}

	private String getRedisKey() {

		return new StringBuilder().append(
			MainConfigLoader.getInstance().getClientCode()).append(
				":").append(getMqKey()).toString();
	}

	abstract protected String getMqKey();

	abstract protected ApplicationContext getContext();

	@SuppressWarnings("unchecked")
	private List<Class<? extends AbstractTask>> getTasks() {

		Tasks tasks = this.getClass().getAnnotation(Tasks.class);
		Class[] classes = tasks.value();
		List<Class<? extends AbstractTask>> taskList = new ArrayList<>();
		for (Class task : classes) {
			taskList.add(task);
		}
		return taskList;
	};

}
